package kafkastudy;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * kafka所用的版本：kafka_2.10-0.10.0.0
 *
 *
 */
public class MyProducer {
    private static final String TOPIC = "my-replicated-topic5"; //kafka创建的topic
    private static final String CONTENT = "This is a single message"; //要发送的内容
    private static final String BROKER_LIST = "10.1.21.5:9092,10.1.21.6:9092,10.1.21.7:9092"; //broker的地址和端口
    private static final String SERIALIZER_CLASS = "kafka.serializer.StringEncoder"; // 序列化类

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties props = new Properties();
        props.put("bootstrap.servers", "10.1.21.5:9092,10.1.21.6:9092,10.1.21.7:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.ByteArraySerializer");

        Producer<byte[], byte[]> producer = new KafkaProducer<>(props);
//        for(int i = 0; i < 100; i++) {
//
//            producer.send(new ProducerRecord<String, String>("xmine", Integer.toString(i), Integer.toString(i)));
//        }
        byte[] key = "key".getBytes();
        byte[] value = "value".getBytes();
        ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[], byte[]>("xmine", key,value);
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                if(e != null)
                    e.printStackTrace();
                System.out.println("The offset of the record we just sent is: " + recordMetadata.offset());
            }
        });

        producer.close();

    }
}